package com.shujia.flink.tf;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.SimpleDateFormat;
import java.time.Duration;

public class Demo06Window {
    public static void main(String[] args) throws Exception {

        // Flink中的窗口操作：时间、计数、会话

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);
        /*
        a,2025/04/12 16:52:51
        a,2025/04/12 16:52:52
        a,2025/04/12 16:52:54
        a,2025/04/12 16:52:55
        a,2025/04/12 16:52:57
         */

        // 预处理数据，解析时间字符串，将其转换成毫秒的时间戳
        SimpleDateFormat format = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");

        SingleOutputStreamOperator<Tuple2<String, Long>> assignDS = socketDS.map(line -> {
                    String[] splits = line.split(",");
                    String word = splits[0];
                    String timeStr = splits[1];
                    long ts = format.parse(timeStr).getTime();
                    return Tuple2.of(word, ts);
                }, Types.TUPLE(Types.STRING, Types.LONG))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((t2, l) -> t2.f1)
                );

//        socketDS
//                .map(w -> Tuple2.of(w, 1), Types.TUPLE(Types.STRING, Types.INT))
//                .keyBy(t2 -> t2.f0)
//                // 基于处理时间的滚动窗口：每隔5s统计单词的数量
////                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                // 基于处理时间的滑动窗口：每隔5s统计最近10s内的单词数量
////                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
//                // 基于处理时间的会话窗口：指定一个时间间隔，例如5s，如果5s没有数据则会触发窗口的统计
////                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
//                // 滚动的计数窗口：对每个Key来说每隔5条数据会统计一次
////                .countWindow(5)
//                // 滑动的计数窗口：对每个Key来说每隔5条数据会统计最近10条数据的一个结果
//                .countWindow(10, 5)
//                .sum(1)
//                .print();

        // 基于事件时间窗口统计
        assignDS.map(t2 -> Tuple2.of(t2.f0, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(kv -> kv.f0)
                // 滚动窗口
//                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 滑动窗口
//                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                // 会话窗口
                .window(EventTimeSessionWindows.withGap(Time.seconds(5)))
                .sum(1)
                .print();


        env.execute();
    }
}
